package com.atguigu.app.dwd.log;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DateFormatUtil_hjy;
import com.atguigu.util.MyKafkaUtil_hjy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author hjy
 * @create 2023/3/24 9:58
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //todo 1 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://47.102.112.46:8020/gmall-flink/check/baseLogApp");
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        env.setStateBackend(new HashMapStateBackend());
//        System.setProperty("HADOOP_USER_NAME","atguigu");
        //todo 2 从kafka（topic_log）获取数据
        String topic="topic_log";
        String groupId="BaseLogApp";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil_hjy.getKafkaConsumer(topic, groupId));

        //todo 3 过滤数据并转为JSONObj对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    try {
                        JSONObject jsonObject = JSONObject.parseObject(value);
                        out.collect(jsonObject);
                    } catch (JSONException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        //todo 4 按照mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(value -> value.getJSONObject("common").getString("mid"));
        //todo 5 新老用户校验
        SingleOutputStreamOperator<JSONObject> isNewDS = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> lastDtState;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last_dt", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                String lastDt = lastDtState.value();
                Long ts = value.getLong("ts");
                JSONObject common = value.getJSONObject("common");
                String isNew = common.getString("is_new");
                String curtDt = DateFormatUtil_hjy.toDate(ts);
                if ("1".equals(isNew)) {
                    if (lastDt == null || lastDt.equals(curtDt)) {
                        lastDtState.update(curtDt);
                    } else {
                        common.put("is_new", "0");
                    }
                } else if (lastDt == null) {
                    lastDtState.update("1970-01-01");
                }
                return value;
            }
        });

        //todo 6 数据分流
        OutputTag<String> startTag = new OutputTag<String>("start") {};
        OutputTag<String> displayTag = new OutputTag<String>("display") {};
        OutputTag<String> actionTag = new OutputTag<String>("action") {};
        OutputTag<String> errorTag = new OutputTag<String>("error") {};
        OutputTag<String> videoTag = new OutputTag<String>("video") {};

        SingleOutputStreamOperator<String> resultDS = isNewDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                JSONObject err = value.getJSONObject("err");
                if (err != null) {
                    ctx.output(errorTag, err.toString());
                    value.remove("err");
                }


                JSONObject start = value.getJSONObject("start");
                JSONObject appVideo = value.getJSONObject("appVideo");
                if (start != null) {
                    ctx.output(startTag, value.toString());
                }else if (appVideo!=null){
                    ctx.output(videoTag,value.toString());
                }
                else {
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        for (Object display : displays) {
                            ctx.output(displayTag, display.toString());
                        }
                    }
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null) {
                        for (Object action : actions) {
                            ctx.output(actionTag, action.toString());
                        }
                    }

                    value.remove("displays");
                    value.remove("actions");
                    JSONObject page = value.getJSONObject("page");
                    if (page!=null){
                        out.collect(value.toJSONString());
                    }
                }
            }
        });

        //todo 7 获取测输出流
        DataStream<String> startSideOutput = resultDS.getSideOutput(startTag);
        DataStream<String> displaySideOutput = resultDS.getSideOutput(displayTag);
        DataStream<String> actionSideOutput = resultDS.getSideOutput(actionTag);
        DataStream<String> errorSideOutput = resultDS.getSideOutput(errorTag);
        DataStream<String> videoSideOutput = resultDS.getSideOutput(videoTag);

        //todo 8 输出
        resultDS.print("resultDS流量域事务事实表 >>>>>>>>>>");


        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String video_topic = "dwd_traffic_video_log";

        resultDS.addSink(MyKafkaUtil_hjy.getKafkaProducer(page_topic));
        startSideOutput.addSink(MyKafkaUtil_hjy.getKafkaProducer(start_topic));
        displaySideOutput.addSink(MyKafkaUtil_hjy.getKafkaProducer(display_topic));
        actionSideOutput.addSink(MyKafkaUtil_hjy.getKafkaProducer(action_topic));
        errorSideOutput.addSink(MyKafkaUtil_hjy.getKafkaProducer(error_topic));
        videoSideOutput.addSink(MyKafkaUtil_hjy.getKafkaProducer(video_topic));

        //todo 9 启动程序
        env.execute("BaseLogApp");

    }
}
